package com.ddone.app.demo;

import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author ddone
 * @date 2024/5/29-13:43
 */
public class FlinkCDCSqlDemoApp {
    public static void main(String[] args) throws Exception {
        // TODO 1. 准备环境
        // 1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        // TODO 2. 检查点
        // 2.1 启用检查点 10秒一个检查的
        env.enableCheckpointing(10 * 1000L);

        // 2.2 设置相邻两次检查点最小间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3 * 1000L);

        // 2.3 设置取消 Job 时检查点的清理模式
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
        );

        // 2.4 设置状态后端类型
        env.setStateBackend(new HashMapStateBackend());

        // 2.5 设置检查点存储路径
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:9820/flink-cdc");

        // 2.6 设置 HDFS 用户名
        System.setProperty("HADOOP_USER_NAME", "bigdata");

        env.setParallelism(1);

        // 1.2 表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2. 创建动态表
        tableEnv.executeSql("CREATE TABLE user_info (\n" +
                "id INT,\n" +
                "name STRING,\n" +
                "age INT,\n" +
                "primary key(id) not enforced\n" +
                ") WITH (" +
                "'connector' = 'mysql-cdc'," +
                "'hostname' = 'hadoop102'," +
                "'port' = '3306'," +
                "'username' = 'root'," +
                "'password' = 'abc123'," +
                "'database-name' = 'gmall_config'," +
                "'table-name' = 'user_info'," +
                "'scan.startup.mode' = 'initial'" +
                ")");

        tableEnv.executeSql("select * from user_info").print();

        // TODO 3. 执行任务
        env.execute();
    }
}
